package com.lefu.risk.storm.rule;

import java.text.SimpleDateFormat;
import java.util.Map;

import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
import org.apache.storm.shade.org.json.simple.JSONObject;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import com.lefu.risk.storm.common.constants.Constant;
import com.lefu.risk.storm.utils.JedisPoolUtil;
import com.lefu.risk.storm.utils.StormCommonUtil;

public class DataRule_01Bolt extends BaseRichBolt{
	private static final long serialVersionUID = 1L;
	private Logger logger;
	private OutputCollector collector;
	private SimpleDateFormat sdf;
	private String key;

	
	@Override
	public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
		this.collector = collector;
		sdf = new SimpleDateFormat("YYYY-MM-dd");
		logger = LogManager.getLogger(DataRule_01Bolt.class);
		
		key = StormCommonUtil.getContextDetil(context);
	}

	@SuppressWarnings("unchecked")
	@Override
	public void execute(Tuple input) {
		JedisPoolUtil redisUtil = null;
		String data =null;
		JSONObject json =null;
		try {
			long a = System.currentTimeMillis();
			
			redisUtil = new JedisPoolUtil(Constant.REDIS_PROPERTIES_NAME);
			data = input.getString(0);
//			Long start = input.getLongByField("start");
			Long start = redisUtil.getMicrosecond()/1000;
			String[] split = data.split("\t");
			json = new JSONObject();
			json.put("ID", split[0]);
			json.put("STATUS", split[1]);
			json.put("AMOUNT", split[2]);
			json.put("CUSTOMER_NO", split[3]);
			json.put("BANK_SHORT", split[4]);
			json.put("CARD_TYPE", split[5]);
			json.put("TRANSATION_TYPE", split[6]);
			json.put("CREATE_TIME", split[7]);
			
			Double amount = null == split[2]?0.00:Double.valueOf(split[2]);
			//logger.info(json);
			this.collector.emit(new Values(split[1],amount,split[5],split[6],split[7],start));
			collector.ack(input);
			redisUtil.lpush(key+"_ACK", json.toJSONString());
			redisUtil.sadd("json_data", json.toJSONString());
		} catch (Exception e) {
			e.printStackTrace();
			collector.fail(input);
			redisUtil.sadd("json_data", json.toJSONString());
			redisUtil.lpush(key+"_FAIL", json.toJSONString());
		}finally{
			if(redisUtil != null) redisUtil.close();
		}
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("STATUS","AMOUNT","CARD_TYPE","TRANSATION_TYPE","TRANSATION_DATE","start"));
	}

}
